﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.ServiceModel;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using GrayParrot.Configuration;
using GrayParrot.RabbitMQ.Configuration;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace GrayParrot.RabbitMQ
{
    public abstract class Subscriber<T> : IDisposable where T : class
    {
        public event EventHandler<Exception> OnConnectionFailed;
        public event EventHandler OnStarted;
        public event EventHandler OnStopped;

        public bool ReadDL { get; set; }
        public string Endpoint { get; set; }
        public string Exchange { get; set; }
        public string RoutingKey { get; set; }
        public string QueueName { get; set; }

        private bool _disposed;
        private static readonly object LockFlag = new object();
        private bool m_ServiceStopRequested = false;
        private string m_userName = string.Empty;
        private string m_password = string.Empty;

        RabbitMQService m_service = null;
        IModel m_channel = null;

        private bool ServiceStopRequested
        {
            get
            {
                lock (LockFlag)
                {
                    return m_ServiceStopRequested;
                }
            }
            set
            {
                lock (LockFlag)
                {
                    m_ServiceStopRequested = value;
                }
            }
        }

        public Subscriber(string subscription)
        {
            ConfigurationManagerHelper pp = new ConfigurationManagerHelper();
            Subscription m_sub = pp.GetSubscription(subscription);
            
            Endpoint = m_sub.ConnectionString;
            Exchange = m_sub.Exchange;
            RoutingKey = m_sub.RoutingKey;
            ReadDL = m_sub.ReadDL;
            QueueName = subscription;

            

            ServiceStopRequested = false;

            //Init();
        }

        /* da usare ?? */
        public Subscriber(Subscription subscription)
        {
            Endpoint = subscription.ConnectionString;
            Exchange = subscription.Exchange;
            RoutingKey = subscription.RoutingKey;
            ReadDL = subscription.ReadDL;
            QueueName = subscription.Name;

            ServiceStopRequested = false;
        }

        public void RequestStop()
        {
            //m_shouldStop = true;
            ServiceStopRequested = true;
        }

        public void Init()
        {
            try
            {
                m_service = new RabbitMQService();
                m_channel = m_service.SetUpSubscriptions(Endpoint);
                //m_service.GetRabbitMQConnection(Endpoint);
                //m_channel = m_service.GetRabbitMQModel();

                if (!ReadDL)
                {
                    //m_currentTopicReceive = TopicReceive;
                    GrayParrot.Logging.Logger.For<T>().Info("Create connection...");
                }
                else
                {
                    //m_currentTopicReceive = TopicReceiveDeadLetter;
                    GrayParrot.Logging.Logger.For<T>().Info("Create connection to DL...");
                }

                if (Endpoint != null)
                {
                    StartListen();
                }
                else
                {
                    GrayParrot.Logging.Logger.For<T>().Error("Connection to Service Broker Failed !");

                    if (OnConnectionFailed != null)
                    {
                        OnConnectionFailed(this, new Exception("Connection to Service Broker Failed !"));
                    }
                }
            }
            catch (Exception ex)
            {
                GrayParrot.Logging.Logger.For<T>().Error("Connection to Service bus Failed !", ex);

                if (OnConnectionFailed != null)
                {
                    OnConnectionFailed(this, ex);
                }
            }
        }

        public void StartListen()
        {
            BasicDeliverEventArgs m_ea = null;

            try
            {
                if (Endpoint != null)
                {
                    if (OnStarted != null)
                    {
                        OnStarted(this, EventArgs.Empty);
                    }

                    GrayParrot.Logging.Logger.For<T>().Info("... connection established");

                    var consumer = new QueueingBasicConsumer(m_channel);
                    m_channel.BasicConsume(QueueName, false, consumer);

                    // Continuously process messages received from the subscription 
                    while (!this.ServiceStopRequested)
                    {
                        try
                        {
                            m_ea = consumer.Queue.Dequeue();

                            Receive(m_ea.Body);

                            m_channel.BasicAck(m_ea.DeliveryTag, false);
                        }
                        catch (Exception ex)
                        {
                            GrayParrot.Logging.Logger.For<T>().Error(ex);

                            m_channel.BasicNack(m_ea.DeliveryTag, false, false);
                        }
                    }
                }
            }
            catch (TimeoutException timeoEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(timeoEx);

                throw;

                //EstablishConnection();
            }
            catch (UnauthorizedAccessException uaEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(uaEx);

                throw;
                //EstablishConnection();
            }
            catch (CommunicationException cmmEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(cmmEx);

                throw;
                //EstablishConnection();
            }
            catch (Exception ex)
            {
                GrayParrot.Logging.Logger.For<T>().Error(ex);

                throw;
                //EstablishConnection();
            }
        }

        //private void SendToDeadLetter(string msg)
        //{
        //    try
        //    {
        //        Publisher m_pub = new Publisher(m_currentSubscription);
        //        m_pub.SendToDeadLetter(msg);
        //        m_pub.Dispose();
        //        m_pub = null;
        //    }
        //    catch (Exception ex)
        //    {
        //        GrayParrot.Logging.Logger.For<T>().Error("Accodamento DL fallito", ex);
        //    }
        //}

        private void EstablishConnection()
        {
            // wait 5 seconds and try to reconnect...
            Thread.Sleep(5000);

            //Endpoint = null;

            // 
            Init();
        }

        public abstract void Receive(byte[] message);

        #region Disposable...
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        ~Subscriber()
        {
            Dispose(false);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_disposed)
                return;

            if (disposing)
            {
                // free other managed objects that implement
                // IDisposable only
            }

            // release any unmanaged objects
            // set the object references to null
            if (m_channel != null)
            {
                m_channel.Close();
            }

            if (m_service != null)
            {
                m_service.Close();
            }

            //
            _disposed = true;
        }
        #endregion
    }
}
